package xjf.learn.stream;

import org.junit.Test;
import xjf.base.constant.STATUS;
import xjf.base.entity.Employee;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * 串行流与并行流
 * <p>
 * ①并行流就是把一个内容分成多个数据块，并用不同额线程分别处理每个数据块的流。多核 CPU 能充分利用
 * <p>
 * ②Java8中将并行进行了优化，我们可以很容易的对数据进行并行操作。Stream API 可以声明性地通过parallel()
 * 与sequential()在并行流与顺序流之间进行切换
 * <p>
 * ③java8的并行流使用到了Fork/Join框架：就是在必要的情况下，将一个大任务，进行拆分成若干个小任务（拆到
 * 不可再拆时），再将一个个的小任务运算的结果进行join汇总。
 * Fork/Join框架采用“工作窃取”模式（work-stealing):当执行新的任务时它可以将其拆分分成更小的任务执行，并将
 * 小任务加到线程队列中，然后再从一个随机线程的队列中偷一个并把它放在自己的队列中。
 * 这种方式减少的线程的等待时间，提高了性能。
 *
 * @Author: xjf
 * @Date: 2020/4/12 22:35
 */
public class Test5 {

    /**
     * 数据准备
     */
    private List<Employee> employeeList = Arrays.asList(
            new Employee("张三", 45, 6000, STATUS.BUSY),
            new Employee("李四", 20, 7600, STATUS.FREE),
            new Employee("王五", 38, 4900, STATUS.VACATION),
            new Employee("赵六", 77, 2000, STATUS.BUSY),
            new Employee("田七", 54, 6500, STATUS.FREE),
            new Employee("田七", 54, 6500, STATUS.FREE)
    );

    /**
     * 1. 并行 与 串行 比较
     */
    @Test
    public void test1() {
        // 串行
        employeeList.stream()
                .forEach(employee -> {
                    System.out.format("foreach: %s [%s]\n",
                            employee.getName(), Thread.currentThread().getName());
                });

        System.out.println("======================================= 分割线 =========================================");

        // 并行
        employeeList.parallelStream()
                .forEach(employee -> {
                    System.out.format("foreach: %s [%s]\n",
                            employee.getName(), Thread.currentThread().getName());
                });
    }

    /**
     * 2. 串行与并行的性能比较： 500万 和 5千万 数字内的偶数求和
     */
    @Test
    public void test2() {
        // 五百万
        int number1 = 500 * 10000;
        // 五千万
        int number2 = 5000 * 10000;


        // 数据准备
        List<Integer> myList = new ArrayList<>();
        for (int i = 0; i < number2; i++) {
            myList.add(i);
        }
        int result = 0;

        // 增强 for 循环
        long loopStartTime = System.currentTimeMillis();
        for (int i : myList) {
            if (i % 2 == 0) {
                result += i;
            }
        }
        long loopEndTime = System.currentTimeMillis();
        System.out.println(result);
        System.out.println("Loop total Time = " + (loopEndTime - loopStartTime));

        System.out.println("======================================= 分割线 =========================================");

        // 串行
        long streamStartTime = System.currentTimeMillis();
        System.out.println(myList.stream().filter(value -> value % 2 == 0).mapToInt(Integer::intValue).sum());
        long streamEndTime = System.currentTimeMillis();
        System.out.println("Stream total Time = " + (streamEndTime - streamStartTime));

        System.out.println("======================================= 分割线 =========================================");

        // 并行
        long parallelStreamStartTime = System.currentTimeMillis();
        System.out.println(myList.parallelStream().filter(value -> value % 2 == 0).mapToInt(Integer::intValue).sum());
        long parallelStreamEndTime = System.currentTimeMillis();
        System.out.println("Parallel Stream total Time = " + (parallelStreamEndTime - parallelStreamStartTime));
    }

    /**
     * 3. 性能测试，专业一些：找出整型数组中的最小值。对比for循环外部迭代和Stream API内部迭代性能。
     *                  求了4次，计算的平均时间
     */
    @Test
    public void test3() {
        // 该方法没有实际意义，热身
        warmUp();

        // 不同的数据长度
        int[] lengths = {
                10000,
                100000,
                1000000,
                10000000,
                100000000,
//                1000000000
        };
        for (int length : lengths) {
            System.out.println();
            System.out.println(String.format("---array length: %d---", length));
            int[] arr = new int[length];
            randomInt(arr);

            int times = 4;
            int min1 = 1;
            int min2 = 2;
            int min3 = 3;
            long startTime;

            // 普通 for 循环
            startTime = System.nanoTime();
            for (int i = 0; i < times; i++) {
                min1 = minIntFor(arr);
            }
            TimeUtil.outTimeUs(startTime, "minIntFor time:", times);

            // 串行
            startTime = System.nanoTime();
            for (int i = 0; i < times; i++) {
                min2 = minIntStream(arr);
            }
            TimeUtil.outTimeUs(startTime, "minIntStream time:", times);

            // 并行
            startTime = System.nanoTime();
            for (int i = 0; i < times; i++) {
                min3 = minIntParallelStream(arr);
            }
            TimeUtil.outTimeUs(startTime, "minIntParallelStream time:", times);

            // 判断计算的结果值是否一致
            System.out.println(min1 == min2 && min2 == min3);
        }
    }


    static class TimeUtil {
        /**
         * 输出时间
         *
         * @param startTime 起始时间
         * @param type      类型字符串
         * @param count     计算次数
         */
        public static void outTimeUs(long startTime, String type, int count) {
            long endTime = System.nanoTime();
            // 用时：(结束时间 - 开始时间) / 次数
            long useTime = (endTime - startTime) / count;

            System.out.println(type + "\t" + useTime);
        }
    }


    private void warmUp() {
        int[] arr = new int[100];
        randomInt(arr);
        for (int i = 0; i < 20000; i++) {
//          minIntFor(arr);
            minIntStream(arr);
            minIntParallelStream(arr);

        }
    }

    private int minIntFor(int[] arr) {
        int min = Integer.MAX_VALUE;
        for (int i = 0; i < arr.length; i++) {
            if (arr[i] < min)
                min = arr[i];
        }
        return min;
    }

    private int minIntStream(int[] arr) {
        return Arrays.stream(arr).min().getAsInt();
    }

    private int minIntParallelStream(int[] arr) {
        return Arrays.stream(arr).parallel().min().getAsInt();
    }

    private void randomInt(int[] arr) {
        Random r = new Random();
        for (int i = 0; i < arr.length; i++) {
            arr[i] = r.nextInt();
        }
    }
}
